package com.weizhu.swufer.core.client;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.ByteString;
import com.weizhu.swufer.constant.SwuferConstant;
import com.weizhu.swufer.core.entity.InvokeParam;
import com.weizhu.swufer.core.exception.RpcBizException;
import com.weizhu.swufer.core.exception.RpcException;
import com.weizhu.swufer.core.filter.Filter;
import com.weizhu.swufer.core.grpc.RemoteCacheManager;
import com.weizhu.swufer.core.metrics.MetricsManager;
import com.weizhu.swufer.core.metrics.RpcMetrics;
import com.weizhu.swufer.core.rpc.*;
import com.weizhu.swufer.core.server.SwuferBootstrap;
import com.weizhu.swufer.core.util.ClassUtil;
import com.weizhu.swufer.core.util.ProtoUtil;
import com.weizhu.swufer.grpc.GrpcCommonServiceGrpc;
import com.weizhu.swufer.grpc.Request;
import com.weizhu.swufer.grpc.Response;
import io.grpc.ManagedChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public abstract class Invoker implements Filter {

    private final static Logger logger = LoggerFactory.getLogger(Invoker.class);

    private static final ListeningExecutorService asyncThreadPool = MoreExecutors.listeningDecorator(MoreExecutors.newDirectExecutorService());

    public boolean isShutdown(RemoteChannel remoteChannel){
        if (remoteChannel == null){
            return true;
        }
        return ((ManagedChannel)remoteChannel.getChannel()).isShutdown();
    }

    /**
     * 调用grpc
     * @param invokeParam
     * @param address
     * @return
     */
     protected Object invokeGrpc(InvokeParam invokeParam, String address)  {
         RpcRequest rpcRequest = invokeParam.getRpcRequest();
         RemoteChannel remoteChannel = RemoteCacheManager.getInstance().getChannel(address);

         RpcMetrics serverRpcMetrics = MetricsManager.getServerRpcMetrics(address);
         serverRpcMetrics.beginStat();
         RpcMetrics methodRpcMetrics = MetricsManager.getMethodRpcMetrics(invokeParam.getServiceMethodName());
         methodRpcMetrics.beginStat();

         List<RpcMetrics> rpcMetrics = new ArrayList<>();
         rpcMetrics.add(serverRpcMetrics);
         rpcMetrics.add(methodRpcMetrics);

         GrpcCommonServiceGrpc.GrpcCommonServiceFutureStub stub = GrpcCommonServiceGrpc
                .newFutureStub(remoteChannel.getChannel());

         RpcResponse rpcResponse;

         Properties properties = SwuferBootstrap.getInstance().getProperties();
         String propertyMaxWaitTime = properties.getProperty(SwuferConstant.PROPERTIES_MAX_WAIT_TIME);
         long maxWaitTime;
         if (propertyMaxWaitTime != null){
             maxWaitTime = Long.parseLong(propertyMaxWaitTime);
         }else {
             maxWaitTime = SwuferConstant.DEFAULT_MAX_WAIT_TIME;
         }
         if (rpcRequest.getMaxWaitTime() > 0){
             maxWaitTime = rpcRequest.getMaxWaitTime();
         }
         if (!isShutdown(remoteChannel)){
             ByteString bytes = ByteString.copyFrom(ProtoUtil.serialize(rpcRequest));
             Request request = Request.newBuilder().setRequest(bytes).build();
             ListenableFuture<Response> grpcListenableFuture = stub.callService(request);
             if (invokeParam.isAsync()){
                 ListenableFuture<?> future = RpcEngine.getInstance().getAsyncThreadPool().submit(new Callable<Object>() {
                     @Override
                     public Object call() throws Exception {
                         return asyncCall(grpcListenableFuture,rpcMetrics);
                     }
                 });
                 RpcContext.getContext().setListenableFuture(future);
                 return null;
             }
             Response response;
             try {
                 response = grpcListenableFuture.get(maxWaitTime, TimeUnit.MILLISECONDS);
             }catch (TimeoutException e){
                 rpcMetrics.forEach(metrics-> metrics.endStat(false));
                 throw new RpcException(RpcException.INVOKE_TIMEOUT,"invoke remote service time out,service:" +
                        rpcRequest.getInterfaceName() + " method:" + rpcRequest.getMethodName() + "address:"
                 +address);
             }catch (Exception e){
                 rpcMetrics.forEach(metrics-> metrics.endStat(false));
                 throw new RpcException(e.getMessage(),e);
             }
             rpcResponse = ProtoUtil.deserialize(response.getResponse().toByteArray(), RpcResponse.class);
         }else{
             RemoteCacheManager.getInstance().removeChannel(address);
             rpcResponse =  RpcResponse.newError("grpc channel is shutdown");
         }

         responseToError(rpcResponse,rpcMetrics);

         return rpcResponse.getData();
    }

    public abstract Object invoke(InvokeParam invokeParam) throws Exception;


     private Object asyncCall(ListenableFuture<Response> grpcListenableFuture,List<RpcMetrics> rpcMetrics){
         RpcResponse rpcResponse;
         Response response;
         try {
              response = grpcListenableFuture.get();
         }catch (Exception e){
             throw new RpcException(e.getMessage(),e);
         }
         rpcResponse = ProtoUtil.deserialize(response.getResponse().toByteArray(), RpcResponse.class);
         responseToError(rpcResponse,rpcMetrics);

         return rpcResponse.getData();
     }

     private void responseToError(RpcResponse rpcResponse,List<RpcMetrics> rpcMetrics){
         if (rpcResponse.getCode() != SwuferConstant.RESPONSE_SUCCESS_CODE){
             if (rpcResponse.getCode() == SwuferConstant.RESPONSE_BIZ_ERROR_CODE){
                 rpcMetrics.forEach(metrics-> metrics.endStat(true));
                 throw new RpcBizException((int)rpcResponse.getData(),rpcResponse.getMsg());
             }
             rpcMetrics.forEach(metrics-> metrics.endStat(false));
             Throwable remoteError = rpcResponse.getError();
             if (remoteError instanceof RpcException){
                 throw (RpcException)remoteError;
             }
             throw new RpcException(rpcResponse.getMsg());
         }
         rpcMetrics.forEach(metrics-> metrics.endStat(true));
     }

    public Object invokeFromMock(RpcRequest rpcRequest){
        String methodName = rpcRequest.getMethodName();
        Object[] args = rpcRequest.getArgs();
        Class<?>[] argTypes = ClassUtil.getParamsType(args);
        Object bean = MockServiceManager.getInstance().getMockServiceMap().get(rpcRequest.getInterfaceName());
        if (bean == null){
            throw new RpcException("no local mock service of " + rpcRequest.getInterfaceName());
        }
        try {
            Method method = null;
            method = bean.getClass().getMethod(methodName,argTypes);
            logger.info("invoke from local mock service:{},method:{} ",bean.getClass().getName() ,method.getName());
            return method.invoke(bean, args);
        } catch (InvocationTargetException e) {
            Throwable trowEx = e.getTargetException();
            throw new RpcBizException("Exception from local mock service : " + trowEx.getMessage(),trowEx);

        } catch (Exception e){
            throw new RpcException(e.getMessage(),e);
        }

    }

}
